import json
import threading
from nanomsg_wrappers import set_wrapper_choice
from nanomsg import SUB,PUB,SUB_SUBSCRIBE,Socket,SUB_UNSUBSCRIBE
from jili.core.printlog import print
import os
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=1)
sendexecutor = ThreadPoolExecutor(max_workers=4)
class SingletonType(type):
    _instance_lock = threading.Lock()
    def __call__(cls):
        if not hasattr(cls, "_instance"):
            with SingletonType._instance_lock:
                if not hasattr(cls, "_instance"):
                    cls._instance = super(SingletonType,cls).__call__()
        return cls._instance
class Counter(metaclass=SingletonType):
    def __init__(self,init=0):
        self.number=init
    def getid(self):
        self.number=self.number+1
        return self.number
counter=Counter()
sourceid=os.getpid()
def subscribe(onfunc="",msgtype="",subject=""):
    s=""
    s0=""
    if subject=="":
        s="{\"0\": \""+msgtype+"\","
        s0="{\"0\":\""+msgtype+"\","
    else:
        s="{\"0\": \""+msgtype+"\", \"1\": \""+subject+"\","
        s0="{\"0\":\""+msgtype+"\",\"1\":\""+subject+"\","
    if isinstance(onfunc,str):
        print(s,s0)
        #print(__onfunclist)
        sub.set_string_option(SUB, SUB_SUBSCRIBE,s)
        sub.set_string_option(SUB, SUB_SUBSCRIBE,s0)
    else:
        if msgtype in __onfunclist.keys():
            if subject in __onfunclist[msgtype].keys():
                if onfunc not in __onfunclist[msgtype][subject]:#去重多次注册函数，否则会多次
                    __onfunclist[msgtype][subject].append(onfunc)
                    print(s,s0)
                    print(__onfunclist)
                    sub.set_string_option(SUB, SUB_SUBSCRIBE,s)
                    sub.set_string_option(SUB, SUB_SUBSCRIBE,s0)
            else:
                __onfunclist[msgtype][subject]=[]
                __onfunclist[msgtype][subject].append(onfunc)
                print(s,s0)
                print(__onfunclist)
                sub.set_string_option(SUB, SUB_SUBSCRIBE,s)
                sub.set_string_option(SUB, SUB_SUBSCRIBE,s0)
        else:
            __onfunclist[msgtype]={}
            __onfunclist[msgtype][subject]=[]
            __onfunclist[msgtype][subject].append(onfunc)
            print(s,s0)
            print(__onfunclist)
            sub.set_string_option(SUB, SUB_SUBSCRIBE,s)
            sub.set_string_option(SUB, SUB_SUBSCRIBE,s0)
def unsubscribe(msgtype="",subject="",onfunc=""):
    s=""
    s0=""
    if subject!="" :
        s="{\"0\": \""+msgtype+"\", \"1\": \""+subject+"\","
        s0="{\"0\":\""+msgtype+"\",\"1\":\""+subject+"\","
        if msgtype in __onfunclist.keys():
            if subject in __onfunclist[msgtype].keys():
                if onfunc=="":
                    del __onfunclist[msgtype][subject]
                    sub.set_string_option(SUB, SUB_UNSUBSCRIBE,s)
                    sub.set_string_option(SUB, SUB_UNSUBSCRIBE,s0)
                else:
                    __onfunclist[msgtype][subject].remove(onfunc)
                    if len(__onfunclist[msgtype][subject])==0:
                        del __onfunclist[msgtype][subject]
                        sub.set_string_option(SUB, SUB_UNSUBSCRIBE,s)
                        sub.set_string_option(SUB, SUB_UNSUBSCRIBE,s0)
    else:
        s="{\"0\": \""+msgtype+"\","
        s0="{\"0\":\""+msgtype+"\","
        if msgtype in __onfunclist.keys():
            if onfunc=="":
                del __onfunclist[msgtype]
                sub.set_string_option(SUB, SUB_UNSUBSCRIBE,s)
                sub.set_string_option(SUB, SUB_UNSUBSCRIBE,s0)
            else:
                if subject in __onfunclist[msgtype].keys():
                    __onfunclist[msgtype][subject].remove(onfunc)
                    if len(__onfunclist[msgtype][subject])==0:
                        del __onfunclist[msgtype][subject]
                        sub.set_string_option(SUB, SUB_UNSUBSCRIBE,s)
                        sub.set_string_option(SUB, SUB_UNSUBSCRIBE,s0)
    print(s,s0)
    print(__onfunclist)
    #sub.set_string_option(SUB, SUB_UNSUBSCRIBE,s)
def getsource(isreply=False):
    t={}
    t["pid"]=sourceid
    t["seqid"]=counter.getid()
    t["isreply"]=isreply
    return t
__addr_sub="tcp://127.0.0.1:8802"
__addr_pub="tcp://127.0.0.1:8801"
pub = Socket(PUB)
__end_send=pub.connect(__addr_pub)
sub = Socket(SUB)
__end_recv = sub.connect(__addr_sub)
__onfunclist={}
__acknotify={}
__acknotify_sycn={}
msgseqkey={}
def isunduplicate(s):
    pid=s["pid"]
    seqid=s["seqid"]
    if pid in msgseqkey.keys():
        if seqid>msgseqkey[pid]:
            return True
        else:
            return False
    else:
        return True
def ontinitmsgseqkey(msg):
    print("init flag",msg)
    pid=msg["s"]["pid"]
    if pid in msgseqkey.keys():
        del msgseqkey[pid]
subscribe(ontinitmsgseqkey,msgtype="initevent",subject="")
def initevent():
    m={}
    m["0"]="initevent"
    m["1"]=""
    m["2"]={}
    m["s"]=getsource()
    try:
        mstr=json.dumps(m)
        pub.send(mstr)
    except Exception as e:
        print(m)
        print("eventlib send:",e)
def onacknotify(msg):
    seqid=msg["s"]["seqid"]
    if __acknotify_sycn:
        if seqid in __acknotify.keys():
            __acknotify[seqid](msg)
        elif seqid in __acknotify_sycn.keys():
            __acknotify_sycn[seqid].update(msg)
            __acknotify_sycn[seqid]["event"].set()
def acksendreply(msg):
    m={}
    m["0"]="acknotify"
    m["1"]=str(msg["s"]["pid"])
    m["2"]={}
    t=msg["s"]
    t["isreply"]=False
    m["s"]=t
    m["e"]=getsource()
    try:
        mstr=json.dumps(m)
        #print(mstr)
        pub.send(mstr)
    except Exception as e:
        print(m)
        print("eventlib send:",e)
subscribe("",msgtype="acknotify",subject=str(sourceid))
def safesend(msgtype,subject,msg,acknotify=""):
    sendexecutor.submit(safesend_sync,msgtype,subject,msg,30,acknotify)
def safesend_sync(msgtype,subject,msg,timeout=30,acknotify=""):
    m={}
    m["0"]=msgtype
    m["1"]=subject
    m["2"]=msg
    m["s"]=getsource(isreply=True)
    if acknotify!="":
        __acknotify[m["s"]["seqid"]]=acknotify
    try:
        mstr=json.dumps(m)
        pub.send(mstr)
    except Exception as e:
        print(m)
        print("eventlib send:",e)
    t=dict(event = threading.Event())
    __acknotify_sycn[m["s"]["seqid"]]=t
    t["event"].clear()
    timeout0=timeout/3
    rst={}
    rst["errcode"]=1
    if t["event"].wait(timeout0):
        if "e" in t.keys():
            rst["errcode"]=0
            rst["info"]=t["e"]
            del __acknotify_sycn[m["s"]["seqid"]]
        else:
            try:
                mstr=json.dumps(m)
                pub.send(mstr)
            except Exception as e:
                print(m)
                print("eventlib send:",e)
            t["event"].clear()
            if t["event"].wait(timeout0):
                if "e" in t.keys():
                    rst["errcode"]=0
                    rst["info"]=t["e"]
                    del __acknotify_sycn[m["s"]["seqid"]]
                else:
                    try:
                        mstr=json.dumps(m)
                        pub.send(mstr)
                    except Exception as e:
                        print(m)
                        print("eventlib send:",e)
                    t["event"].clear()
                    if t["event"].wait(timeout0):
                        if "e" in t.keys():
                            rst["errcode"]=0
                            rst["info"]=t["e"]
                            del __acknotify_sycn[m["s"]["seqid"]]
    if rst["errcode"]==1:
        print("消息超时：",m)
    return rst
def send(msgtype,subject,msg):
    m={}
    m["0"]=msgtype
    m["1"]=subject
    m["2"]=msg
    m["s"]=getsource()
    try:
        mstr=json.dumps(m)
        pub.send(mstr)
    except Exception as e:
        print(m)
        print("eventlib send:",e)
    return m["s"]
def getsubcall():
    return __onfunclist
def setmax_workers(maxwokers):
    global executor 
    executor = ThreadPoolExecutor(max_workers=maxwokers)
def __deal():
    while 1:
        try:
            msg = sub.recv()
            msg=str(msg,encoding="utf-8")
            #print("recv:",msg)
            msg=json.loads(msg)
            msgtype=msg["0"]
            subject=msg["1"]
            if msgtype=="acknotify":
                pid=msg["s"]["pid"]
                if pid == sourceid:
                    executor.submit(onacknotify, msg)
                    if msgtype in __onfunclist.keys():
                        if "" in __onfunclist[msgtype].keys():
                            for call in __onfunclist[msgtype][""]:
                                executor.submit(call, msg)
                        if (subject!="") & (subject in __onfunclist[msgtype].keys()):
                            for call in __onfunclist[msgtype][subject]:
                                executor.submit(call, msg)
            else:
                if msgtype in __onfunclist.keys():
                    if "s" in msg.keys():
                        s=msg["s"]
                        if s["isreply"]:
                            acksendreply(msg)
                            if isunduplicate(s):
                                if "" in __onfunclist[msgtype].keys():
                                    for call in __onfunclist[msgtype][""]:
                                        executor.submit(call, msg)
                                if (subject!="") & (subject in __onfunclist[msgtype].keys()):
                                    for call in __onfunclist[msgtype][subject]:
                                        executor.submit(call, msg)
                        else:
                            if "" in __onfunclist[msgtype].keys():
                                for call in __onfunclist[msgtype][""]:
                                    executor.submit(call, msg)
                            if (subject!="") & (subject in __onfunclist[msgtype].keys()):
                                for call in __onfunclist[msgtype][subject]:
                                    executor.submit(call, msg)
                    else:
                        if "" in __onfunclist[msgtype].keys():
                            for call in __onfunclist[msgtype][""]:
                                executor.submit(call, msg)
                        if (subject!="") & (subject in __onfunclist[msgtype].keys()):
                            for call in __onfunclist[msgtype][subject]:
                                executor.submit(call, msg)
        except Exception as e:
            print("eventlib __deal:",e)
initevent()
t1 = threading.Thread(target=__deal,args=())
t1.start()
print(t1.getName())